diff --git a/PROGRESS.md b/PROGRESS.md index 66329e5..5cb8826 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -44,6 +44,14 @@ tests, `-race` in `make test`. auto-tuner (outcome ratio → bounded cooldown, persisted as `source=feedback`). - **Rules:** water/meal/break (sev1–2 care), service_down (sev4, `poll:uptimekuma`), 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:` 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 on a real CRITICAL); kuma monitor_status → `service_down`. Writes only on value-change (no append-only churn). @@ -260,13 +268,17 @@ Capability-class gaps — built but thin: "clarify". 7. **Presence is effectively one signal** (page_heartbeat); desk_active is 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/ - memory` (jul6 task 7 + follow-up) has a `Store` interface + in-memory cosine - impl; notes **and facts** are indexed on capture, and `IntentQuery` now reads - it back (after notes-RAG misses, before general-knowledge) — fact recall - («когда я пил воду?») is its distinct payoff. Still missing: a persistent/ - external vector backend (the store is lost on restart) — the read seam is - there to swap onto one. Persona prompt and custom TTS voice (kami-picked, +8. **Long-term memory is now persistent (store-backed), not the spec's chroma.** + `internal/memory` has a `Store` interface; the daemon now wires + `store.MemoryStore` (`internal/store/memory.go`) — a **persistent** backend + in the **same encrypted sqlite db** (survives restarts; recall text inherits + at-rest encryption, so no plaintext sidecar). Vectors are float32 blobs, + search is brute-force cosine (fine at single-user scale; ANN is the later + 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. Ops footnote: in the Docker deploy, voice-over-web needs mavend to bind its diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 50649ef..d4e6ea0 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -128,7 +128,7 @@ func run(args []string) error { defer phr.Close() // ----- 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 { return fmt.Errorf("wire voice: %w", err) } @@ -175,7 +175,7 @@ func run(args []string) error { tickInterval := time.Duration(cfg.TickInterval) repeatInterval := time.Duration(cfg.RepeatInterval) 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) ----- coreAPI := &daemonAPI{ diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go index e7c0aa0..1ce653a 100644 --- a/cmd/mavend/tick.go +++ b/cmd/mavend/tick.go @@ -23,6 +23,7 @@ import ( "github.com/kami/maven/internal/ipc" "github.com/kami/maven/internal/loop" "github.com/kami/maven/internal/phraser" + "github.com/kami/maven/internal/routine" "github.com/kami/maven/internal/store" ) @@ -53,6 +54,12 @@ type tickLoop struct { // immediately (legacy behaviour). 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. // populated when digestCfg != nil && digestCfg.Enabled. digestQ []QueuedNudge @@ -76,6 +83,7 @@ func newTickLoop( rules []loop.Rule, tickInterval, repeatInterval, autotuneInterval time.Duration, digestCfg *config.DigestConfig, + routines []routine.Routine, ) *tickLoop { return &tickLoop{ store: st, @@ -88,6 +96,8 @@ func newTickLoop( autotuneInterval: autotuneInterval, digestCfg: digestCfg, digestQ: nil, + routines: routines, + routineLast: make(map[string]time.Time), 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. 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 // delivery. a failed send leaves the reminder pending — the next tick // 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 } +// 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 // (autotuneInterval, see run) so it doesn't write a fact every tick. for each // rule: diff --git a/cmd/mavend/tick_test.go b/cmd/mavend/tick_test.go index 795b09e..367eab7 100644 --- a/cmd/mavend/tick_test.go +++ b/cmd/mavend/tick_test.go @@ -10,6 +10,7 @@ import ( "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" ) @@ -45,7 +46,54 @@ func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink, digestCf Nudges: 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. diff --git a/cmd/mavend/voice.go b/cmd/mavend/voice.go index 7d5e6d1..4a00f4d 100644 --- a/cmd/mavend/voice.go +++ b/cmd/mavend/voice.go @@ -111,7 +111,7 @@ func (w *voiceWiring) close() { // // When voice is enabled, MUST wire a voicesink into the dispatcher's Voice // 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 { 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) ----- w.voiceSink = voicesink.New(synthesizer, sessions) - // ----- memory (long-term vector storage, in-memory for now) ----- - memStore := memory.NewInMemoryStore() + // ----- memory (long-term vector storage) ----- + // 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) ----- dialogueSessions := dialogue.NewSessionStore(2 * time.Minute) diff --git a/internal/config/config.go b/internal/config/config.go index a6f4941..783150b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,7 @@ import ( "github.com/kami/maven/internal/delivery/ntfysink" "github.com/kami/maven/internal/delivery/telegramsink" + "github.com/robfig/cron/v3" ) // 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 // (every nudge is sent as it fires — legacy behaviour). 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 @@ -363,6 +383,14 @@ func (c *Config) applyDefaults() { 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 { @@ -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 } diff --git a/internal/routine/routine.go b/internal/routine/routine.go new file mode 100644 index 0000000..7e883c4 --- /dev/null +++ b/internal/routine/routine.go @@ -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 +} diff --git a/internal/routine/routine_test.go b/internal/routine/routine_test.go new file mode 100644 index 0000000..8e1550a --- /dev/null +++ b/internal/routine/routine_test.go @@ -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) + } + }) +} diff --git a/internal/store/memory.go b/internal/store/memory.go new file mode 100644 index 0000000..f2b43b1 --- /dev/null +++ b/internal/store/memory.go @@ -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 +} diff --git a/internal/store/memory_test.go b/internal/store/memory_test.go new file mode 100644 index 0000000..3e57243 --- /dev/null +++ b/internal/store/memory_test.go @@ -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) + } +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 01925d0..7bfd2af 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -20,6 +20,12 @@ var migrations = []string{ `ALTER TABLE tools ADD COLUMN scope TEXT NOT NULL DEFAULT 'homelab';`, // #1 `ALTER TABLE reminders ADD COLUMN cron TEXT; 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