From 92ecb691de84f40f203f7742bc6087e2297dd1c8 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 31 Jul 2026 13:50:43 +0400 Subject: [PATCH] Re-embed stored notes and facts after an embedder swap (#378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The embedder swap left every stored vector in the old model's space, so cosine against a new query vector is noise. Add the one-shot backfill: store.ReembedAll re-embeds every note and fact text with the currently configured embedder (the passage side, which is the side stored text was written with) and rewrites both places a vector lives — the notes table embedding column and the memory_vectors rows. All of it plus the embedder marker happens in one transaction, so a failure partway changes nothing and writes no marker: re-run it. A run against a DB whose marker already names the current embedder does nothing. Triggered explicitly with `mavend -reembed`, not automatically on mismatch: ONNX on the laptop CPU makes this minutes of work, and a silent multi-minute stall on boot would look like a hang. The mismatch warning now tells the user to run it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ --- cmd/mavend/main.go | 2 + cmd/mavend/voice.go | 42 +++++++- internal/store/backfill.go | 161 ++++++++++++++++++++++++++++++ internal/store/backfill_test.go | 171 ++++++++++++++++++++++++++++++++ internal/store/meta.go | 6 +- 5 files changed, 376 insertions(+), 6 deletions(-) create mode 100644 internal/store/backfill.go create mode 100644 internal/store/backfill_test.go diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 94bb51c..bc436c4 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -189,7 +189,9 @@ func (l *lockedAPI) MorningStatus(ctx context.Context) ([]ipc.MorningRoutineStat func run(args []string) error { cfgPath := flag.String("config", defaultConfigPath(), "path to mavend JSON config") wrappedKeyPath := flag.String("wrapped-key-file", "", "path to wrapped encryption key blob (enables cold-start unlock)") + reembed := flag.Bool("reembed", false, "re-embed every stored note and fact with the configured embedder, then serve normally (run once after an embedder swap)") flag.CommandLine.Parse(args) + reembedOnStart = *reembed cfg, err := config.Load(*cfgPath) if err != nil { return err diff --git a/cmd/mavend/voice.go b/cmd/mavend/voice.go index 06a1b35..25f5e24 100644 --- a/cmd/mavend/voice.go +++ b/cmd/mavend/voice.go @@ -1769,26 +1769,62 @@ func jsonStringImpl(s string) string { return string(b) } +// reembedOnStart is the -reembed flag (set in run()). Opt-in on purpose: see +// runReembed. +var reembedOnStart bool + // checkStoredEmbedder compares the embedder we just loaded with the one that // wrote the vectors already in the DB (Vikunja #378). // // The two models we have both make 384-dim vectors, so a size check catches // nothing: after a swap, recall silently compares vectors from different -// spaces and the scores are noise. So we say it out loud. Nothing is changed -// here — recall keeps running exactly as before until the backfill lands. +// spaces and the scores are noise. So we say it out loud. Recall itself is not +// changed here — the fix is `mavend -reembed`. func checkStoredEmbedder(dataStore *store.Store, emb router.Embedder) { if dataStore == nil { return } current := router.EmbedderID(emb) + if reembedOnStart { + runReembed(dataStore, emb, current) + return + } stored, mismatch, err := dataStore.CheckEmbedder(context.Background(), current) if err != nil { log.Printf("voice: embedder marker check failed: %v", err) return } if mismatch { - log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded (Vikunja #378)", stored, current) + log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded — run `mavend -reembed` once (Vikunja #378)", stored, current) return } log.Printf("voice: embedder marker ok (%s)", current) } + +// runReembed is the one-shot backfill behind -reembed. +// +// Why a flag and not automatic on mismatch: the embedder is ONNX on the +// laptop's CPU, so a few thousand notes is minutes of work. Doing that silently +// inside a normal start would look like the daemon hanging on boot. So the user +// runs it once, deliberately, after an embedder swap; the mismatch warning +// above tells them to. It re-embeds, logs what it did, and then the daemon +// carries on serving as usual — no separate binary, no second start needed. +func runReembed(dataStore *store.Store, emb router.Embedder, current string) { + log.Printf("voice: re-embedding stored notes and facts with %s — this can take a few minutes, do not interrupt", current) + res, err := dataStore.ReembedAll(context.Background(), current, + // EmbedPassage, not EmbedQuery: these are stored texts being searched + // FOR, which is the side they were written with. + func(ctx context.Context, text string) ([]float32, error) { + return router.EmbedPassage(ctx, emb, text) + }) + if err != nil { + log.Printf("voice: re-embed FAILED, nothing was changed and no marker was written — safe to run again: %v", err) + return + } + if res.Skipped { + log.Printf("voice: re-embed skipped — the stored vectors were already written by %s", current) + return + } + log.Printf("voice: re-embed done — %d notes in the notes table, %d notes and %d facts in the memory index, %d rows had no text to re-embed, took %s; stored vectors now belong to %s", + res.Notes, res.MemNotes, res.Facts, res.NoText, res.Took.Round(time.Second), current) +} diff --git a/internal/store/backfill.go b/internal/store/backfill.go new file mode 100644 index 0000000..2de783d --- /dev/null +++ b/internal/store/backfill.go @@ -0,0 +1,161 @@ +package store + +import ( + "context" + "encoding/json" + "fmt" + "time" +) + +// EmbedFunc embeds one piece of stored text. The caller passes +// router.EmbedPassage — the STORE side of the query/passage asymmetry, which is +// the side every vector in the DB was written with. (Passing the query side +// would put the stored vectors in the wrong half of the space and quietly halve +// recall.) A func instead of an interface keeps this package free of any +// dependency on internal/router. +type EmbedFunc func(ctx context.Context, text string) ([]float32, error) + +// BackfillResult is what the re-embed run did, for logging. +type BackfillResult struct { + Skipped bool // marker already matched — nothing to do + Notes int // rows rewritten in the notes table + Facts int // fact rows rewritten in memory_vectors + MemNotes int // note rows rewritten in memory_vectors + NoText int // memory_vectors rows with no text in their meta, left alone + Took time.Duration +} + +// ReembedAll rewrites every stored vector with the currently configured +// embedder and then records that embedder as the one that owns the DB. +// +// Both places a vector lives are rewritten in the same pass: the `notes` table +// `embedding` column and the `memory_vectors` rows (notes AND facts). Doing +// only one would leave the two indexes disagreeing, which is worse than leaving +// both stale. +// +// Safe to re-run: if the marker already names the current embedder there is +// nothing to fix, so it returns immediately with Skipped set. +// +// Crash safety: everything — every vector and the marker — happens inside one +// transaction. If anything fails or the process dies partway, the transaction +// rolls back: no vectors changed and no marker written, so the next run does +// the whole job again. The marker is never set unless the full rewrite +// committed. +func (s *Store) ReembedAll(ctx context.Context, currentID string, embed EmbedFunc) (BackfillResult, error) { + start := time.Now() + var res BackfillResult + + stored, err := s.Meta(ctx, metaKeyEmbedderID) + if err != nil { + return res, err + } + if stored == currentID { + res.Skipped = true + res.Took = time.Since(start) + return res, nil + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return res, fmt.Errorf("reembed: begin: %w", err) + } + defer tx.Rollback() // no-op once committed + + // ----- notes table ----- + type noteRow struct { + id int64 + text string + } + var notes []noteRow + rows, err := tx.QueryContext(ctx, `SELECT id, text FROM notes WHERE text != ''`) + if err != nil { + return res, fmt.Errorf("reembed: read notes: %w", err) + } + for rows.Next() { + var n noteRow + if err := rows.Scan(&n.id, &n.text); err != nil { + rows.Close() + return res, fmt.Errorf("reembed: note row: %w", err) + } + notes = append(notes, n) + } + rows.Close() + if err := rows.Err(); err != nil { + return res, fmt.Errorf("reembed: notes: %w", err) + } + + for _, n := range notes { + vec, err := embed(ctx, n.text) + if err != nil { + return res, fmt.Errorf("reembed: embed note %d: %w", n.id, err) + } + if _, err := tx.ExecContext(ctx, + `UPDATE notes SET embedding = ? WHERE id = ?`, floatsToBlob(vec), n.id); err != nil { + return res, fmt.Errorf("reembed: write note %d: %w", n.id, err) + } + res.Notes++ + } + + // ----- memory_vectors (the unified index: notes AND facts) ----- + // The text to re-embed is the one carried in the row's meta blob, which is + // exactly the text that was embedded when the row was written. + type vecRow struct { + id, text, kind string + } + var vecs []vecRow + rows, err = tx.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`) + if err != nil { + return res, fmt.Errorf("reembed: read memory vectors: %w", err) + } + for rows.Next() { + var id, metaJSON string + if err := rows.Scan(&id, &metaJSON); err != nil { + rows.Close() + return res, fmt.Errorf("reembed: memory row: %w", err) + } + meta := map[string]string{} + if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil { + rows.Close() + return res, fmt.Errorf("reembed: meta for %q: %w", id, err) + } + if meta["text"] == "" { + res.NoText++ + continue + } + vecs = append(vecs, vecRow{id: id, text: meta["text"], kind: meta["type"]}) + } + rows.Close() + if err := rows.Err(); err != nil { + return res, fmt.Errorf("reembed: memory vectors: %w", err) + } + + for _, v := range vecs { + vec, err := embed(ctx, v.text) + if err != nil { + return res, fmt.Errorf("reembed: embed %q: %w", v.id, err) + } + if _, err := tx.ExecContext(ctx, + `UPDATE memory_vectors SET vec = ? WHERE id = ?`, encodeVec(vec), v.id); err != nil { + return res, fmt.Errorf("reembed: write %q: %w", v.id, err) + } + if v.kind == "fact" { + res.Facts++ + } else { + res.MemNotes++ + } + } + + // Same transaction as the rewrite, on purpose: the marker can only exist if + // every vector above was written. + if _, err := tx.ExecContext(ctx, + `INSERT INTO meta (key, value) VALUES (?,?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + metaKeyEmbedderID, currentID); err != nil { + return res, fmt.Errorf("reembed: write marker: %w", err) + } + if err := tx.Commit(); err != nil { + return res, fmt.Errorf("reembed: commit: %w", err) + } + res.Took = time.Since(start) + return res, nil +} diff --git a/internal/store/backfill_test.go b/internal/store/backfill_test.go new file mode 100644 index 0000000..01cf00d --- /dev/null +++ b/internal/store/backfill_test.go @@ -0,0 +1,171 @@ +package store + +import ( + "context" + "errors" + "testing" + "time" +) + +// markerVec is a recognisable vector: nothing in these tests writes it except +// the backfill, so finding it proves the row really was rewritten. +var markerVec = []float32{9, 9, 9} + +func newEmbedder(calls *int) EmbedFunc { + return func(_ context.Context, _ string) ([]float32, error) { + *calls++ + return markerVec, nil + } +} + +// seedOldVectors puts one note (notes table + unified index) and one fact +// (unified index only) in the DB, both carrying obviously-old vectors. +func seedOldVectors(t *testing.T, s *Store) { + t.Helper() + ctx := context.Background() + old := []float32{0.1, 0.2, 0.3} + id, err := s.WriteNote(ctx, time.Now(), "молоко в холодильнике", old, "voice") + if err != nil { + t.Fatalf("WriteNote: %v", err) + } + mem := s.VectorMemory() + if err := mem.Insert(ctx, "note:1", old, map[string]string{ + "type": "note", "text": "молоко в холодильнике", + }); err != nil { + t.Fatalf("Insert note vector: %v", err) + } + if err := mem.Insert(ctx, "fact:water:1", old, map[string]string{ + "type": "fact", "text": "я пил воду", + }); err != nil { + t.Fatalf("Insert fact vector: %v", err) + } + _ = id +} + +func noteVec(t *testing.T, s *Store) []float32 { + t.Helper() + var blob []byte + if err := s.db.QueryRow(`SELECT embedding FROM notes LIMIT 1`).Scan(&blob); err != nil { + t.Fatalf("read note embedding: %v", err) + } + return blobToFloats(blob) +} + +func memVec(t *testing.T, s *Store, id string) []float32 { + t.Helper() + var blob []byte + if err := s.db.QueryRow(`SELECT vec FROM memory_vectors WHERE id = ?`, id).Scan(&blob); err != nil { + t.Fatalf("read memory vector %s: %v", id, err) + } + return decodeVec(blob) +} + +func sameVec(a, b []float32) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// The deployed case: old vectors everywhere, no marker. Every vector in both +// places must be rewritten and the marker recorded. +func TestReembedAllRewritesEveryVector(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedOldVectors(t, s) + + calls := 0 + res, err := s.ReembedAll(ctx, "multilingual-e5-small@384", newEmbedder(&calls)) + if err != nil { + t.Fatalf("ReembedAll: %v", err) + } + if res.Skipped { + t.Fatal("first run should not skip") + } + if res.Notes != 1 || res.MemNotes != 1 || res.Facts != 1 { + t.Fatalf("counts: notes=%d memNotes=%d facts=%d", res.Notes, res.MemNotes, res.Facts) + } + if calls != 3 { + t.Fatalf("embedder called %d times, want 3", calls) + } + if !sameVec(noteVec(t, s), markerVec) { + t.Fatalf("notes table not rewritten: %v", noteVec(t, s)) + } + if !sameVec(memVec(t, s, "note:1"), markerVec) { + t.Fatal("unified index note row not rewritten") + } + if !sameVec(memVec(t, s, "fact:water:1"), markerVec) { + t.Fatal("unified index fact row not rewritten") + } + got, err := s.Meta(ctx, metaKeyEmbedderID) + if err != nil { + t.Fatalf("Meta: %v", err) + } + if got != "multilingual-e5-small@384" { + t.Fatalf("marker = %q", got) + } +} + +// Re-running must do nothing at all — not a second pass over the same rows. +func TestReembedAllSecondRunIsNoop(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedOldVectors(t, s) + + calls := 0 + if _, err := s.ReembedAll(ctx, "e5@384", newEmbedder(&calls)); err != nil { + t.Fatalf("first run: %v", err) + } + first := calls + + res, err := s.ReembedAll(ctx, "e5@384", newEmbedder(&calls)) + if err != nil { + t.Fatalf("second run: %v", err) + } + if !res.Skipped { + t.Fatal("second run should report Skipped") + } + if calls != first { + t.Fatalf("second run embedded %d more rows, want 0", calls-first) + } +} + +// A failure partway must leave the DB exactly as it was: no marker, and the old +// vectors still in place (one transaction, rolled back). +func TestReembedAllPartialFailureLeavesMarkerUnset(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedOldVectors(t, s) + before := noteVec(t, s) + + calls := 0 + boom := func(_ context.Context, _ string) ([]float32, error) { + calls++ + if calls == 2 { + return nil, errors.New("onnx blew up") + } + return markerVec, nil + } + if _, err := s.ReembedAll(ctx, "e5@384", boom); err == nil { + t.Fatal("expected an error") + } + got, err := s.Meta(ctx, metaKeyEmbedderID) + if err != nil { + t.Fatalf("Meta: %v", err) + } + if got != "" { + t.Fatalf("marker was set to %q after a failed run", got) + } + if !sameVec(noteVec(t, s), before) { + t.Fatal("a failed run left a partially rewritten notes table") + } + // And the mismatch warning must still fire, so the user knows to re-run. + if _, mismatch, err := s.CheckEmbedder(ctx, "e5@384"); err != nil || !mismatch { + t.Fatalf("CheckEmbedder after failed backfill: mismatch=%v err=%v", mismatch, err) + } +} diff --git a/internal/store/meta.go b/internal/store/meta.go index 8edd0e3..652110f 100644 --- a/internal/store/meta.go +++ b/internal/store/meta.go @@ -60,9 +60,9 @@ const EmbedderUnknown = "unknown (written before this marker existed)" // marker was added to catch. // - marker absent and no vectors ⇒ fresh DB, claim it, nothing to fix. // -// TODO(#378): on a mismatch, run the one-shot backfill here — re-embed every -// stored note and fact text with the current embedder (EmbedPassage side), -// write the vectors back, then SetMeta the current id. +// On a mismatch the fix is ReembedAll (backfill.go), run explicitly with +// `mavend -reembed`. Nothing is re-embedded here: that work is minutes of CPU +// on the laptop and must not stall a normal start. func (s *Store) CheckEmbedder(ctx context.Context, currentID string) (stored string, mismatch bool, err error) { stored, err = s.Meta(ctx, metaKeyEmbedderID) if err != nil {