maven: scheduled routines + persistent long-term memory
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
This commit is contained in:
+19
-7
@@ -44,6 +44,14 @@ tests, `-race` in `make test`.
|
|||||||
auto-tuner (outcome ratio → bounded cooldown, persisted as `source=feedback`).
|
auto-tuner (outcome ratio → bounded cooldown, persisted as `source=feedback`).
|
||||||
- **Rules:** water/meal/break (sev1–2 care), service_down (sev4, `poll:uptimekuma`),
|
- **Rules:** water/meal/break (sev1–2 care), service_down (sev4, `poll:uptimekuma`),
|
||||||
netdata_critical (sev3, `poll:netdata`).
|
netdata_critical (sev3, `poll:netdata`).
|
||||||
|
- **Routines (`internal/routine`):** operator-declared clockwork — the third
|
||||||
|
proactive class beside reminders (user-stated) and care rules (world-state).
|
||||||
|
Config `routines[]` (cron + literal RU body + severity) fire through the normal
|
||||||
|
dispatcher on schedule (an 08:00 briefing, a 22:00 wind-down). Bodies are
|
||||||
|
literal (not LLM-phrased ⇒ can't hallucinate); rule name `routine:<name>` so
|
||||||
|
they don't pollute the care autotuner; cold-start guard seeds on first sight so
|
||||||
|
a restart never replays a missed schedule. Pure `routine.Due`, unit-tested; the
|
||||||
|
tick driver holds the last-fired map.
|
||||||
- **Env facts (`mavpoll`):** netdata alarms → `netdata_alarm` (fires immediately
|
- **Env facts (`mavpoll`):** netdata alarms → `netdata_alarm` (fires immediately
|
||||||
on a real CRITICAL); kuma monitor_status → `service_down`. Writes only on
|
on a real CRITICAL); kuma monitor_status → `service_down`. Writes only on
|
||||||
value-change (no append-only churn).
|
value-change (no append-only churn).
|
||||||
@@ -260,13 +268,17 @@ Capability-class gaps — built but thin:
|
|||||||
"clarify".
|
"clarify".
|
||||||
7. **Presence is effectively one signal** (page_heartbeat); desk_active is
|
7. **Presence is effectively one signal** (page_heartbeat); desk_active is
|
||||||
still an undeployed script — "voice when near" routing runs on a guess.
|
still an undeployed script — "voice when near" routing runs on a guess.
|
||||||
8. **Long-term memory is in-memory only, not the spec's chroma.** `internal/
|
8. **Long-term memory is now persistent (store-backed), not the spec's chroma.**
|
||||||
memory` (jul6 task 7 + follow-up) has a `Store` interface + in-memory cosine
|
`internal/memory` has a `Store` interface; the daemon now wires
|
||||||
impl; notes **and facts** are indexed on capture, and `IntentQuery` now reads
|
`store.MemoryStore` (`internal/store/memory.go`) — a **persistent** backend
|
||||||
it back (after notes-RAG misses, before general-knowledge) — fact recall
|
in the **same encrypted sqlite db** (survives restarts; recall text inherits
|
||||||
(«когда я пил воду?») is its distinct payoff. Still missing: a persistent/
|
at-rest encryption, so no plaintext sidecar). Vectors are float32 blobs,
|
||||||
external vector backend (the store is lost on restart) — the read seam is
|
search is brute-force cosine (fine at single-user scale; ANN is the later
|
||||||
there to swap onto one. Persona prompt and custom TTS voice (kami-picked,
|
swap behind the same interface). Notes **and facts** are indexed on capture;
|
||||||
|
`IntentQuery` reads it back (after notes-RAG misses, before general-knowledge)
|
||||||
|
— fact recall («когда я пил воду?») is its distinct payoff. The in-memory
|
||||||
|
impl remains the test/no-store floor. Remaining: an ANN/external index is
|
||||||
|
optional-scale, not a gap. Persona prompt and custom TTS voice (kami-picked,
|
||||||
replaces the irina floor — [[custom-voice-training]]) are still future items.
|
replaces the irina floor — [[custom-voice-training]]) are still future items.
|
||||||
|
|
||||||
Ops footnote: in the Docker deploy, voice-over-web needs mavend to bind its
|
Ops footnote: in the Docker deploy, voice-over-web needs mavend to bind its
|
||||||
|
|||||||
+2
-2
@@ -128,7 +128,7 @@ func run(args []string) error {
|
|||||||
defer phr.Close()
|
defer phr.Close()
|
||||||
|
|
||||||
// ----- voice: reactive audio path (TCP listener + stt/router/tts) -----
|
// ----- voice: reactive audio path (TCP listener + stt/router/tts) -----
|
||||||
voiceW, err := wireVoice(cfg, ipc.NewStoreAPI(st), phr)
|
voiceW, err := wireVoice(cfg, ipc.NewStoreAPI(st), phr, st.VectorMemory())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("wire voice: %w", err)
|
return fmt.Errorf("wire voice: %w", err)
|
||||||
}
|
}
|
||||||
@@ -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)
|
||||||
tl := newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest)
|
tl := newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines))
|
||||||
|
|
||||||
// ----- IPC boundary (core ↔ modules) -----
|
// ----- IPC boundary (core ↔ modules) -----
|
||||||
coreAPI := &daemonAPI{
|
coreAPI := &daemonAPI{
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ import (
|
|||||||
"github.com/kami/maven/internal/ipc"
|
"github.com/kami/maven/internal/ipc"
|
||||||
"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/routine"
|
||||||
"github.com/kami/maven/internal/store"
|
"github.com/kami/maven/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -53,6 +54,12 @@ type tickLoop struct {
|
|||||||
// immediately (legacy behaviour).
|
// immediately (legacy behaviour).
|
||||||
digestCfg *config.DigestConfig
|
digestCfg *config.DigestConfig
|
||||||
|
|
||||||
|
// routines — operator-declared scheduled behaviors. fired through the
|
||||||
|
// dispatcher when their cron crosses. routineLast tracks the per-routine
|
||||||
|
// last-fire time across ticks (the driver owns it; routine.Due mutates it).
|
||||||
|
routines []routine.Routine
|
||||||
|
routineLast map[string]time.Time
|
||||||
|
|
||||||
// digestQ — in-memory queue of eligible nudges waiting for batch flush.
|
// digestQ — in-memory queue of eligible nudges waiting for batch flush.
|
||||||
// populated when digestCfg != nil && digestCfg.Enabled.
|
// populated when digestCfg != nil && digestCfg.Enabled.
|
||||||
digestQ []QueuedNudge
|
digestQ []QueuedNudge
|
||||||
@@ -76,6 +83,7 @@ func newTickLoop(
|
|||||||
rules []loop.Rule,
|
rules []loop.Rule,
|
||||||
tickInterval, repeatInterval, autotuneInterval time.Duration,
|
tickInterval, repeatInterval, autotuneInterval time.Duration,
|
||||||
digestCfg *config.DigestConfig,
|
digestCfg *config.DigestConfig,
|
||||||
|
routines []routine.Routine,
|
||||||
) *tickLoop {
|
) *tickLoop {
|
||||||
return &tickLoop{
|
return &tickLoop{
|
||||||
store: st,
|
store: st,
|
||||||
@@ -88,6 +96,8 @@ func newTickLoop(
|
|||||||
autotuneInterval: autotuneInterval,
|
autotuneInterval: autotuneInterval,
|
||||||
digestCfg: digestCfg,
|
digestCfg: digestCfg,
|
||||||
digestQ: nil,
|
digestQ: nil,
|
||||||
|
routines: routines,
|
||||||
|
routineLast: make(map[string]time.Time),
|
||||||
lastPhrase: make(map[string]delivery.PhrasedNudge),
|
lastPhrase: make(map[string]delivery.PhrasedNudge),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -160,6 +170,12 @@ func (t *tickLoop) tick(ctx context.Context, now time.Time) {
|
|||||||
// (with dedup) avoids re-queueing the same rule after a flush.
|
// (with dedup) avoids re-queueing the same rule after a flush.
|
||||||
t.maybeFlush(ctx, now, state)
|
t.maybeFlush(ctx, now, state)
|
||||||
|
|
||||||
|
// routines: operator-declared scheduled behaviors. fire the ones whose cron
|
||||||
|
// crossed since last fire, delivered through the normal routing (voice when
|
||||||
|
// present, away channels otherwise). bodies are literal operator text — not
|
||||||
|
// LLM-phrased — so a routine can't hallucinate. severity comes from config.
|
||||||
|
t.fireRoutines(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.
|
||||||
@@ -307,6 +323,45 @@ func (t *tickLoop) flushDigest(ctx context.Context, now time.Time, state loop.St
|
|||||||
t.digestQ = nil
|
t.digestQ = nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// routinesFromConfig maps the config's routine blocks to the engine type.
|
||||||
|
// Validation (cron parses, name/body present, severity defaulted) already ran
|
||||||
|
// in config.Load, so this is a pure field copy.
|
||||||
|
func routinesFromConfig(rc []config.RoutineConfig) []routine.Routine {
|
||||||
|
if len(rc) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
out := make([]routine.Routine, len(rc))
|
||||||
|
for i, r := range rc {
|
||||||
|
out[i] = routine.Routine{Name: r.Name, Cron: r.Cron, Body: r.Body, Severity: r.Severity}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// fireRoutines dispatches the routines whose cron schedule crossed since their
|
||||||
|
// last fire. Each is delivered as a nudge through the normal routing table
|
||||||
|
// (ChannelsFor(severity, presence)) with a "routine:"-prefixed rule name so it
|
||||||
|
// can't collide with a care rule in the feedback autotuner. A dispatch failure
|
||||||
|
// logs and continues — one bad send must not skip the rest, and routine.Due has
|
||||||
|
// already advanced the last-fire time so a transient failure drops that fire
|
||||||
|
// rather than replaying it every tick (a routine is clockwork, not an alarm —
|
||||||
|
// no repeat-til-ack).
|
||||||
|
func (t *tickLoop) fireRoutines(ctx context.Context, now time.Time, state loop.State) {
|
||||||
|
for _, r := range routine.Due(t.routines, t.routineLast, now) {
|
||||||
|
pn := delivery.PhrasedNudge{
|
||||||
|
Candidate: loop.Candidate{
|
||||||
|
Rule: loop.Rule{Name: "routine:" + r.Name, Severity: loop.Severity(r.Severity)},
|
||||||
|
Severity: loop.Severity(r.Severity),
|
||||||
|
State: state,
|
||||||
|
},
|
||||||
|
Body: r.Body,
|
||||||
|
Summary: r.Body,
|
||||||
|
}
|
||||||
|
if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil {
|
||||||
|
log.Printf("tick: dispatch routine %s: %v", r.Name, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// 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:
|
||||||
|
|||||||
+49
-1
@@ -10,6 +10,7 @@ import (
|
|||||||
"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/routine"
|
||||||
"github.com/kami/maven/internal/store"
|
"github.com/kami/maven/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -45,7 +46,54 @@ func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink, digestCf
|
|||||||
Nudges: st,
|
Nudges: st,
|
||||||
Reminders: st,
|
Reminders: st,
|
||||||
})
|
})
|
||||||
return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0, digestCfg)
|
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.
|
// refNow — fixed tick time so presence decay + since durations are deterministic.
|
||||||
|
|||||||
+7
-3
@@ -111,7 +111,7 @@ func (w *voiceWiring) close() {
|
|||||||
//
|
//
|
||||||
// When voice is enabled, MUST wire a voicesink into the dispatcher's Voice
|
// When voice is enabled, MUST wire a voicesink into the dispatcher's Voice
|
||||||
// slot using w.sessions (the caller does that — see main.go).
|
// slot using w.sessions (the caller does that — see main.go).
|
||||||
func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser) (*voiceWiring, error) {
|
func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, memStore memory.Store) (*voiceWiring, error) {
|
||||||
if cfg.Voice == nil || !cfg.Voice.Enabled {
|
if cfg.Voice == nil || !cfg.Voice.Enabled {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
@@ -203,8 +203,12 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser) (*v
|
|||||||
// ----- voice sink (proactive nudges: dispatcher → voicesink → tts → push to client) -----
|
// ----- voice sink (proactive nudges: dispatcher → voicesink → tts → push to client) -----
|
||||||
w.voiceSink = voicesink.New(synthesizer, sessions)
|
w.voiceSink = voicesink.New(synthesizer, sessions)
|
||||||
|
|
||||||
// ----- memory (long-term vector storage, in-memory for now) -----
|
// ----- memory (long-term vector storage) -----
|
||||||
memStore := memory.NewInMemoryStore()
|
// Persistent (store-backed, survives restarts) when the daemon passes one;
|
||||||
|
// falls back to the in-memory floor otherwise (tests / no-store paths).
|
||||||
|
if memStore == nil {
|
||||||
|
memStore = memory.NewInMemoryStore()
|
||||||
|
}
|
||||||
|
|
||||||
// ----- dialogue (multi-turn slot carry-over; 2-min follow-up window) -----
|
// ----- dialogue (multi-turn slot carry-over; 2-min follow-up window) -----
|
||||||
dialogueSessions := dialogue.NewSessionStore(2 * time.Minute)
|
dialogueSessions := dialogue.NewSessionStore(2 * time.Minute)
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ import (
|
|||||||
|
|
||||||
"github.com/kami/maven/internal/delivery/ntfysink"
|
"github.com/kami/maven/internal/delivery/ntfysink"
|
||||||
"github.com/kami/maven/internal/delivery/telegramsink"
|
"github.com/kami/maven/internal/delivery/telegramsink"
|
||||||
|
"github.com/robfig/cron/v3"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Config — the daemon's whole config tree. Loaded once at startup.
|
// Config — the daemon's whole config tree. Loaded once at startup.
|
||||||
@@ -119,6 +120,25 @@ type Config struct {
|
|||||||
// Digest — notification batching / digest mode. nil ⇒ digest disabled
|
// Digest — notification batching / digest mode. nil ⇒ digest disabled
|
||||||
// (every nudge is sent as it fires — legacy behaviour).
|
// (every nudge is sent as it fires — legacy behaviour).
|
||||||
Digest *DigestConfig `json:"digest,omitempty"`
|
Digest *DigestConfig `json:"digest,omitempty"`
|
||||||
|
|
||||||
|
// Routines — scheduled behaviors maven performs on a cron schedule (a
|
||||||
|
// morning briefing, an evening wind-down), independent of any request or
|
||||||
|
// care predicate. Each fires its Body through the dispatcher on its Cron
|
||||||
|
// schedule. Empty ⇒ no routines. See internal/routine for the class
|
||||||
|
// distinction from reminders (user-stated) and care rules (world-state).
|
||||||
|
Routines []RoutineConfig `json:"routines,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// RoutineConfig — one scheduled routine. Cron is a standard 5-field expression
|
||||||
|
// ("0 8 * * *" = 08:00 daily). Body is the RU text delivered verbatim (routines
|
||||||
|
// are not LLM-phrased). Severity (1-4, default 1) drives routing: care-class
|
||||||
|
// (≤2) is suppressed by quiet hours and drops when away; ops-class reaches away
|
||||||
|
// channels.
|
||||||
|
type RoutineConfig struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
Cron string `json:"cron"`
|
||||||
|
Body string `json:"body"`
|
||||||
|
Severity int `json:"severity,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// QuietHoursConfig — a recurring daily quiet-window. Times are local to the
|
// QuietHoursConfig — a recurring daily quiet-window. Times are local to the
|
||||||
@@ -363,6 +383,14 @@ func (c *Config) applyDefaults() {
|
|||||||
c.Voice.ToolTimeout = Duration(DefaultToolTimeout)
|
c.Voice.ToolTimeout = Duration(DefaultToolTimeout)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// routines: default severity to care-class (1) — the safe floor: a
|
||||||
|
// misconfigured routine can't blast an away channel at 3am.
|
||||||
|
for i := range c.Routines {
|
||||||
|
if c.Routines[i].Severity == 0 {
|
||||||
|
c.Routines[i].Severity = 1
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Config) validate() error {
|
func (c *Config) validate() error {
|
||||||
@@ -382,6 +410,19 @@ func (c *Config) validate() error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// routines: name + body required, cron must parse. A typo here should fail
|
||||||
|
// at startup, not silently never fire.
|
||||||
|
for _, r := range c.Routines {
|
||||||
|
if r.Name == "" {
|
||||||
|
return errors.New("routine: name is required")
|
||||||
|
}
|
||||||
|
if r.Body == "" {
|
||||||
|
return fmt.Errorf("routine %q: body is required", r.Name)
|
||||||
|
}
|
||||||
|
if _, err := cron.ParseStandard(r.Cron); err != nil {
|
||||||
|
return fmt.Errorf("routine %q: bad cron %q: %w", r.Name, r.Cron, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,89 @@
|
|||||||
|
// Package routine is maven's scheduled-behavior engine: things she does on a
|
||||||
|
// cron schedule, independent of any reactive request or care predicate.
|
||||||
|
//
|
||||||
|
// A Routine is the third proactive class, distinct from the two that already
|
||||||
|
// exist:
|
||||||
|
//
|
||||||
|
// - a Reminder (internal/store) is a USER-stated one-off/recurring intent —
|
||||||
|
// "напомни завтра позвонить маме". It exists because the user asked.
|
||||||
|
// - a care Rule (internal/loop) fires on WORLD-STATE predicates — water/meal/
|
||||||
|
// break/service_down. It exists because a snapshot crossed a threshold.
|
||||||
|
// - a Routine is OPERATOR-declared clockwork — an 08:00 morning briefing, a
|
||||||
|
// 22:00 wind-down. It exists purely because the clock said so.
|
||||||
|
//
|
||||||
|
// This package is pure: no store, no clock of its own, no I/O. The daemon's tick
|
||||||
|
// driver owns the impurity (it holds the last-fired map and calls Due each tick),
|
||||||
|
// exactly as it does for the loop package. That keeps the schedule logic here
|
||||||
|
// unit-testable without time side effects.
|
||||||
|
package routine
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/robfig/cron/v3"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Routine — one scheduled behavior. Cron is a standard 5-field expression
|
||||||
|
// (minute hour dom month dow). Body is the RU text delivered verbatim through
|
||||||
|
// the normal dispatcher (routines are NOT LLM-phrased: the operator writes the
|
||||||
|
// line, so it's deterministic and can't hallucinate). Severity (1-4) drives
|
||||||
|
// routing the same way a nudge's does — care-class (≤2) is suppressed by quiet
|
||||||
|
// hours and drops when away; ops-class (3-4) reaches away channels.
|
||||||
|
type Routine struct {
|
||||||
|
Name string
|
||||||
|
Cron string
|
||||||
|
Body string
|
||||||
|
Severity int
|
||||||
|
}
|
||||||
|
|
||||||
|
// Validate reports the first structural problem with a routine set: a missing
|
||||||
|
// name/cron/body or an unparseable cron expression. Called at config load so a
|
||||||
|
// typo surfaces at startup, not as a silently-never-firing routine at runtime.
|
||||||
|
func Validate(routines []Routine) error {
|
||||||
|
for _, r := range routines {
|
||||||
|
if r.Name == "" {
|
||||||
|
return fmt.Errorf("routine: name is required")
|
||||||
|
}
|
||||||
|
if r.Body == "" {
|
||||||
|
return fmt.Errorf("routine %q: body is required", r.Name)
|
||||||
|
}
|
||||||
|
if _, err := cron.ParseStandard(r.Cron); err != nil {
|
||||||
|
return fmt.Errorf("routine %q: bad cron %q: %w", r.Name, r.Cron, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Due returns the routines whose schedule crossed since their last fire and
|
||||||
|
// records now as the new last-fire time for each one returned. The caller owns
|
||||||
|
// `last` (the tick driver holds it across ticks); Due mutates it in place.
|
||||||
|
//
|
||||||
|
// Cold-start guard: a routine absent from `last` (never seen — fresh daemon, or
|
||||||
|
// a newly-added routine) is SEEDED to now WITHOUT firing. Without this, a daemon
|
||||||
|
// restart at 12:00 would replay the 08:00 briefing, because Next() computed from
|
||||||
|
// the zero time is always in the past. The cost is that a routine can't fire in
|
||||||
|
// the same tick the daemon booted — an acceptable trade for never replaying a
|
||||||
|
// missed schedule on restart.
|
||||||
|
//
|
||||||
|
// A routine whose cron fails to parse is skipped (Validate rejects those at
|
||||||
|
// load; this is defence in depth so a bad expression can't fire every tick).
|
||||||
|
func Due(routines []Routine, last map[string]time.Time, now time.Time) []Routine {
|
||||||
|
var out []Routine
|
||||||
|
for _, r := range routines {
|
||||||
|
sched, err := cron.ParseStandard(r.Cron)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
prev, seen := last[r.Name]
|
||||||
|
if !seen {
|
||||||
|
last[r.Name] = now // seed on first sight — don't fire (cold-start guard)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if next := sched.Next(prev); !next.After(now) {
|
||||||
|
last[r.Name] = now
|
||||||
|
out = append(out, r)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
@@ -0,0 +1,88 @@
|
|||||||
|
package routine
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestValidate(t *testing.T) {
|
||||||
|
ok := []Routine{{Name: "morning", Cron: "0 8 * * *", Body: "доброе утро"}}
|
||||||
|
if err := Validate(ok); err != nil {
|
||||||
|
t.Fatalf("valid routine rejected: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
r Routine
|
||||||
|
}{
|
||||||
|
{"missing name", Routine{Cron: "0 8 * * *", Body: "x"}},
|
||||||
|
{"missing body", Routine{Name: "m", Cron: "0 8 * * *"}},
|
||||||
|
{"bad cron", Routine{Name: "m", Cron: "not a cron", Body: "x"}},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
t.Run(c.name, func(t *testing.T) {
|
||||||
|
if err := Validate([]Routine{c.r}); err == nil {
|
||||||
|
t.Error("expected an error, got nil")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDue(t *testing.T) {
|
||||||
|
// a routine that fires at 08:00 every day.
|
||||||
|
rs := []Routine{{Name: "morning", Cron: "0 8 * * *", Body: "доброе утро", Severity: 1}}
|
||||||
|
|
||||||
|
t.Run("first sight seeds without firing (cold-start guard)", func(t *testing.T) {
|
||||||
|
last := map[string]time.Time{}
|
||||||
|
now := time.Date(2026, 7, 6, 8, 0, 0, 0, time.UTC) // exactly on schedule
|
||||||
|
got := Due(rs, last, now)
|
||||||
|
if len(got) != 0 {
|
||||||
|
t.Errorf("routine fired on first sight: %v", got)
|
||||||
|
}
|
||||||
|
if _, seen := last["morning"]; !seen {
|
||||||
|
t.Error("first sight did not seed the last-fired map")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("fires once the schedule crosses", func(t *testing.T) {
|
||||||
|
last := map[string]time.Time{"morning": time.Date(2026, 7, 6, 7, 30, 0, 0, time.UTC)}
|
||||||
|
now := time.Date(2026, 7, 6, 8, 0, 30, 0, time.UTC) // just past 08:00
|
||||||
|
got := Due(rs, last, now)
|
||||||
|
if len(got) != 1 || got[0].Name != "morning" {
|
||||||
|
t.Fatalf("expected morning to fire, got %v", got)
|
||||||
|
}
|
||||||
|
if !last["morning"].Equal(now) {
|
||||||
|
t.Errorf("last-fired not advanced to now: %v", last["morning"])
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("does not refire before the next crossing", func(t *testing.T) {
|
||||||
|
last := map[string]time.Time{"morning": time.Date(2026, 7, 6, 8, 0, 0, 0, time.UTC)}
|
||||||
|
now := time.Date(2026, 7, 6, 8, 5, 0, 0, time.UTC) // same morning, later
|
||||||
|
if got := Due(rs, last, now); len(got) != 0 {
|
||||||
|
t.Errorf("routine refired within the same window: %v", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("missed schedule on restart fires at most once, not per-tick", func(t *testing.T) {
|
||||||
|
// booted at 07:00, seeded then; next tick is 09:00 (08:00 already passed).
|
||||||
|
last := map[string]time.Time{"morning": time.Date(2026, 7, 6, 7, 0, 0, 0, time.UTC)}
|
||||||
|
now := time.Date(2026, 7, 6, 9, 0, 0, 0, time.UTC)
|
||||||
|
if got := Due(rs, last, now); len(got) != 1 {
|
||||||
|
t.Fatalf("missed 08:00 should fire once at 09:00, got %v", got)
|
||||||
|
}
|
||||||
|
// immediately after, it must not fire again.
|
||||||
|
if got := Due(rs, last, now.Add(time.Minute)); len(got) != 0 {
|
||||||
|
t.Errorf("routine fired twice for one missed schedule: %v", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("bad cron is skipped, not fired every tick", func(t *testing.T) {
|
||||||
|
bad := []Routine{{Name: "broken", Cron: "nonsense", Body: "x"}}
|
||||||
|
last := map[string]time.Time{"broken": time.Date(2026, 7, 6, 7, 0, 0, 0, time.UTC)}
|
||||||
|
now := time.Date(2026, 7, 6, 9, 0, 0, 0, time.UTC)
|
||||||
|
if got := Due(bad, last, now); len(got) != 0 {
|
||||||
|
t.Errorf("unparseable cron fired: %v", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -0,0 +1,131 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"encoding/binary"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"math"
|
||||||
|
"sort"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/memory"
|
||||||
|
)
|
||||||
|
|
||||||
|
// MemoryStore is the persistent backend for internal/memory's vector Store,
|
||||||
|
// sharing the main encrypted sqlite database so recall text (note/fact bodies
|
||||||
|
// carried in the meta blob) inherits at-rest encryption — a plaintext sidecar
|
||||||
|
// file would undercut store.OpenEncrypted. It survives daemon restarts, which
|
||||||
|
// the InMemoryStore does not: that was the last gap keeping long-term memory
|
||||||
|
// from being real.
|
||||||
|
//
|
||||||
|
// Search is brute-force cosine over every row loaded into memory — the same
|
||||||
|
// algorithm as InMemoryStore, just sourced from disk. At the single-user note+
|
||||||
|
// fact scale (thousands of rows, not millions) a full scan per query is well
|
||||||
|
// under a millisecond; an ANN index is the swap for later, behind this same
|
||||||
|
// interface. Vectors are assumed L2-normalized by the embedder, so cosine is a
|
||||||
|
// dot product.
|
||||||
|
type MemoryStore struct {
|
||||||
|
db *sql.DB
|
||||||
|
}
|
||||||
|
|
||||||
|
// VectorMemory returns a persistent memory.Store backed by this store's db.
|
||||||
|
// The returned store shares the db handle (single writer — the daemon), so it
|
||||||
|
// participates in the same encrypted tmpfs working copy and is sealed on Close.
|
||||||
|
func (s *Store) VectorMemory() *MemoryStore {
|
||||||
|
return &MemoryStore{db: s.db}
|
||||||
|
}
|
||||||
|
|
||||||
|
// compile-time check: MemoryStore satisfies the memory.Store interface.
|
||||||
|
var _ memory.Store = (*MemoryStore)(nil)
|
||||||
|
|
||||||
|
// Insert upserts a vector by id: a repeated id replaces the prior row rather
|
||||||
|
// than accumulating duplicates (the note/fact ids are stable and unique, so a
|
||||||
|
// re-index is an update, not a second copy — an improvement on InMemoryStore's
|
||||||
|
// append-always). meta is stored as a JSON object.
|
||||||
|
func (m *MemoryStore) Insert(ctx context.Context, id string, vec []float32, meta map[string]string) error {
|
||||||
|
metaJSON, err := json.Marshal(meta)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("memory: marshal meta: %w", err)
|
||||||
|
}
|
||||||
|
_, err = m.db.ExecContext(ctx,
|
||||||
|
`INSERT INTO memory_vectors (id, vec, meta, created_ts) VALUES (?,?,?,?)
|
||||||
|
ON CONFLICT(id) DO UPDATE SET vec = excluded.vec, meta = excluded.meta, created_ts = excluded.created_ts`,
|
||||||
|
id, encodeVec(vec), string(metaJSON), time.Now().UnixMilli())
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("memory: insert %q: %w", id, err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Search returns the topK nearest rows by cosine similarity. A full scan; see
|
||||||
|
// the type doc for why that's fine at this scale.
|
||||||
|
func (m *MemoryStore) Search(ctx context.Context, vec []float32, topK int) ([]memory.Result, error) {
|
||||||
|
if topK <= 0 {
|
||||||
|
topK = 10
|
||||||
|
}
|
||||||
|
rows, err := m.db.QueryContext(ctx, `SELECT id, vec, meta FROM memory_vectors`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("memory: scan: %w", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var out []memory.Result
|
||||||
|
for rows.Next() {
|
||||||
|
var id, metaJSON string
|
||||||
|
var blob []byte
|
||||||
|
if err := rows.Scan(&id, &blob, &metaJSON); err != nil {
|
||||||
|
return nil, fmt.Errorf("memory: row: %w", err)
|
||||||
|
}
|
||||||
|
meta := map[string]string{}
|
||||||
|
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
|
||||||
|
return nil, fmt.Errorf("memory: unmarshal meta for %q: %w", id, err)
|
||||||
|
}
|
||||||
|
out = append(out, memory.Result{ID: id, Score: dot(vec, decodeVec(blob)), Meta: meta})
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, fmt.Errorf("memory: rows: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
sort.Slice(out, func(i, j int) bool { return out[i].Score > out[j].Score })
|
||||||
|
if topK < len(out) {
|
||||||
|
out = out[:topK]
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// encodeVec serializes a float32 slice as little-endian IEEE-754 bytes (4 bytes
|
||||||
|
// per element) for the BLOB column.
|
||||||
|
func encodeVec(v []float32) []byte {
|
||||||
|
b := make([]byte, 4*len(v))
|
||||||
|
for i, f := range v {
|
||||||
|
binary.LittleEndian.PutUint32(b[4*i:], math.Float32bits(f))
|
||||||
|
}
|
||||||
|
return b
|
||||||
|
}
|
||||||
|
|
||||||
|
// decodeVec reverses encodeVec. A blob whose length isn't a multiple of 4 is
|
||||||
|
// truncated to the whole-element prefix (defensive — a well-formed row can't
|
||||||
|
// produce that).
|
||||||
|
func decodeVec(b []byte) []float32 {
|
||||||
|
n := len(b) / 4
|
||||||
|
v := make([]float32, n)
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
|
v[i] = math.Float32frombits(binary.LittleEndian.Uint32(b[4*i:]))
|
||||||
|
}
|
||||||
|
return v
|
||||||
|
}
|
||||||
|
|
||||||
|
// dot is the cosine similarity for L2-normalized vectors (mismatched lengths ⇒
|
||||||
|
// 0, matching internal/memory's cosine).
|
||||||
|
func dot(a, b []float32) float64 {
|
||||||
|
if len(a) != len(b) || len(a) == 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
var sum float64
|
||||||
|
for i := range a {
|
||||||
|
sum += float64(a[i]) * float64(b[i])
|
||||||
|
}
|
||||||
|
return sum
|
||||||
|
}
|
||||||
@@ -0,0 +1,103 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func newMemTestStore(t *testing.T) *Store {
|
||||||
|
t.Helper()
|
||||||
|
path := filepath.Join(t.TempDir(), "mem_test.db")
|
||||||
|
st, err := Open(context.Background(), path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Open: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = st.Close() })
|
||||||
|
return st
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMemoryStoreInsertSearch(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
m := newMemTestStore(t).VectorMemory()
|
||||||
|
|
||||||
|
// three orthonormal-ish vectors; a query aligned with the second must rank it top.
|
||||||
|
if err := m.Insert(ctx, "a", []float32{1, 0, 0}, map[string]string{"text": "вода"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := m.Insert(ctx, "b", []float32{0, 1, 0}, map[string]string{"text": "сон", "type": "fact"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := m.Insert(ctx, "c", []float32{0, 0, 1}, map[string]string{"text": "еда"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
got, err := m.Search(ctx, []float32{0, 1, 0}, 2)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Search: %v", err)
|
||||||
|
}
|
||||||
|
if len(got) != 2 {
|
||||||
|
t.Fatalf("topK=2 returned %d results", len(got))
|
||||||
|
}
|
||||||
|
if got[0].ID != "b" {
|
||||||
|
t.Errorf("top hit = %q, want b", got[0].ID)
|
||||||
|
}
|
||||||
|
if got[0].Meta["text"] != "сон" || got[0].Meta["type"] != "fact" {
|
||||||
|
t.Errorf("meta not round-tripped: %v", got[0].Meta)
|
||||||
|
}
|
||||||
|
if got[0].Score < 0.99 {
|
||||||
|
t.Errorf("aligned vector score = %v, want ~1.0", got[0].Score)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMemoryStoreUpsertReplaces(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
m := newMemTestStore(t).VectorMemory()
|
||||||
|
|
||||||
|
if err := m.Insert(ctx, "x", []float32{1, 0}, map[string]string{"text": "старое"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := m.Insert(ctx, "x", []float32{0, 1}, map[string]string{"text": "новое"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
got, err := m.Search(ctx, []float32{0, 1}, 10)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != 1 {
|
||||||
|
t.Fatalf("re-inserting the same id produced %d rows, want 1 (upsert)", len(got))
|
||||||
|
}
|
||||||
|
if got[0].Meta["text"] != "новое" {
|
||||||
|
t.Errorf("upsert kept the old value: %q", got[0].Meta["text"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMemoryStorePersistsAcrossReopen(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
path := filepath.Join(t.TempDir(), "persist.db")
|
||||||
|
|
||||||
|
st, err := Open(ctx, path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := st.VectorMemory().Insert(ctx, "k", []float32{1, 0, 0}, map[string]string{"text": "запомни"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := st.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// reopen the same file — the in-memory floor would have lost this.
|
||||||
|
st2, err := Open(ctx, path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = st2.Close() })
|
||||||
|
got, err := st2.VectorMemory().Search(ctx, []float32{1, 0, 0}, 1)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != 1 || got[0].Meta["text"] != "запомни" {
|
||||||
|
t.Fatalf("memory did not survive reopen: %v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -20,6 +20,12 @@ var migrations = []string{
|
|||||||
`ALTER TABLE tools ADD COLUMN scope TEXT NOT NULL DEFAULT 'homelab';`, // #1
|
`ALTER TABLE tools ADD COLUMN scope TEXT NOT NULL DEFAULT 'homelab';`, // #1
|
||||||
`ALTER TABLE reminders ADD COLUMN cron TEXT;
|
`ALTER TABLE reminders ADD COLUMN cron TEXT;
|
||||||
ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
||||||
|
`CREATE TABLE memory_vectors (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
vec BLOB NOT NULL,
|
||||||
|
meta TEXT NOT NULL DEFAULT '{}',
|
||||||
|
created_ts INTEGER NOT NULL
|
||||||
|
);`, // #3 — long-term vector memory (persistent backend for internal/memory)
|
||||||
}
|
}
|
||||||
|
|
||||||
// migrate applies every migration with a number greater than the DB's current
|
// migrate applies every migration with a number greater than the DB's current
|
||||||
|
|||||||
Reference in New Issue
Block a user