kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge the memory and store sweep (#244)

Browse Source 
The embedder prefix audit came back clean, which was the one finding worth
escalating. Every EmbedQuery, EmbedPassage and Embed call site across
internal/store, internal/memory and their cmd/mavend callers agrees. No naked
Embed on a note.

ReembedAll and RepairFactVectors each ran an identical select and scan over
memory_vectors before diverging on what to do with the row. One
allMemVectorMetas now, parameterized over a small interface so it serves
backfill's transaction and factvectors' plain read alike.

Two swallowed errors. AcceptProposedRoutine read RowsAffected with a discarded
error where every other call in the same file checks it, so a driver error read
as zero rows. MarkAcked did the same, and the branch it fed was dead, since both
arms returned nil. The swallowed error and the branch went together.

The agent refuted the rest of the brief. Repeated scans and swallowed errors
were one instance each rather than the pattern tasks.go showed. Both packages
carry per-type scan helpers already, and every magic value is already named with
its reason beside it, which reads as the residue of earlier sweep waves.

(V-581)
This commit is contained in:
claude
2026-08-06 03:15:10 +04:00
parent 69270f4cfb 5bca435146
commit 95cbf82e38
5 changed files with 62 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/store/ack.go
+4 -6
View File
@@ -55,17 +55,15 @@ func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) {
// alarm (voice acknowledgment, Telegram callback, etc.).
func (s *Store) MarkAcked(ctx context.Context, key string) error {
now := time.Now()
res, err := s.db.ExecContext(ctx,
// No pending nudges is not an error — already acked or never sent — so the
// rows-affected count is not read at all: every outcome below this line is
// the same nil.
_, err := s.db.ExecContext(ctx,
`UPDATE nudges SET outcome = 'acted', outcome_ts = ?
WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`,
now.UnixMilli(), key)
if err != nil {
return fmt.Errorf("mark acked %s: %w", key, err)
}
n, _ := res.RowsAffected()
if n == 0 {
// no pending nudges — already acked or never sent; not an error.
return nil
}
return nil
}
internal/store/backfill.go
+5 -20
View File
@@ -2,7 +2,6 @@ package store
import (
"context"
"encoding/json"
"fmt"
"time"
)
@@ -103,30 +102,16 @@ func (s *Store) ReembedAll(ctx context.Context, currentID string, embed EmbedFun
id, text, kind string
}
var vecs []vecRow
rows, err = tx.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
memRows, err := allMemVectorMetas(ctx, tx)
if err != nil {
return res, fmt.Errorf("reembed: read memory vectors: %w", err)
return res, fmt.Errorf("reembed: %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"] == "" {
for _, v := range memRows {
if v.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)
vecs = append(vecs, vecRow{id: v.ID, text: v.Meta["text"], kind: v.Meta["type"]})
}
for _, v := range vecs {
internal/store/factvectors.go
+6 -20
View File
@@ -64,9 +64,9 @@ func (s *Store) RepairFactVectors(ctx context.Context, embed EmbedFunc) (FactVec
return res, nil
}
rows, err := s.db.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
memRows, err := allMemVectorMetas(ctx, s.db)
if err != nil {
return res, fmt.Errorf("repair fact vectors: read: %w", err)
return res, fmt.Errorf("repair fact vectors: %w", err)
}
type factVec struct {
id, key string
@@ -75,33 +75,19 @@ func (s *Store) RepairFactVectors(ctx context.Context, embed EmbedFunc) (FactVec
}
var vecs []factVec
newest := map[string]int64{} // key → newest ts seen for it
for rows.Next() {
var id, metaJSON string
if err := rows.Scan(&id, &metaJSON); err != nil {
rows.Close()
return res, fmt.Errorf("repair fact vectors: row: %w", err)
}
meta := map[string]string{}
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
rows.Close()
return res, fmt.Errorf("repair fact vectors: meta for %q: %w", id, err)
}
if meta["type"] != "fact" {
for _, v := range memRows {
if v.Meta["type"] != "fact" {
continue
}
key, ts, ok := splitFactVectorID(id)
key, ts, ok := splitFactVectorID(v.ID)
if !ok {
continue
}
vecs = append(vecs, factVec{id: id, key: key, meta: meta, ts: ts})
vecs = append(vecs, factVec{id: v.ID, key: key, meta: v.Meta, ts: ts})
if ts > newest[key] {
newest[key] = ts
}
}
rows.Close()
if err := rows.Err(); err != nil {
return res, fmt.Errorf("repair fact vectors: rows: %w", err)
}
for _, v := range vecs {
drop := v.ts < newest[v.key]
internal/store/memory.go
+43
View File
@@ -169,6 +169,49 @@ func (m *MemoryStore) DeletePrefix(ctx context.Context, prefix string) (int64, e
return n, nil
}
// memVectorRow is one memory_vectors row with its meta blob decoded — the
// shape both ReembedAll (backfill.go) and RepairFactVectors (factvectors.go)
// read the whole table as, before each decides what to do with a row on its
// own terms (one keys off meta["text"], the other off meta["type"] and the
// id's embedded key/timestamp). Query-then-scan was duplicated across the two
// before this, id-for-id.
type memVectorRow struct {
ID string
Meta map[string]string
}
// queryContexter is the common surface *sql.DB and *sql.Tx share that
// allMemVectorMetas needs. ReembedAll reads inside a transaction so its
// migration is atomic; RepairFactVectors reads directly off the db handle.
type queryContexter interface {
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}
// allMemVectorMetas reads every memory_vectors row and decodes its meta blob.
func allMemVectorMetas(ctx context.Context, q queryContexter) ([]memVectorRow, error) {
rows, err := q.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
if err != nil {
return nil, fmt.Errorf("read memory vectors: %w", err)
}
defer rows.Close()
var out []memVectorRow
for rows.Next() {
var id, metaJSON string
if err := rows.Scan(&id, &metaJSON); err != nil {
return nil, fmt.Errorf("memory vector row: %w", err)
}
meta := map[string]string{}
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
return nil, fmt.Errorf("meta for %q: %w", id, err)
}
out = append(out, memVectorRow{ID: id, Meta: meta})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("memory vectors: %w", err)
}
return out, nil
}
// escapeLike neutralises the LIKE wildcards in a literal prefix.
func escapeLike(s string) string {
r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
internal/store/proposed_routines.go
+4 -1
View File
@@ -187,7 +187,10 @@ func (s *Store) AcceptProposedRoutine(ctx context.Context, id int64, ts time.Tim
if err != nil {
return fmt.Errorf("accept proposed routine: %w", err)
}
n, _ := res.RowsAffected()
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("accept proposed routine: rows affected: %w", err)
}
if n == 0 {
return fmt.Errorf("%w: id=%d not in 'proposed' status", ErrProposedRoutineNotFound, id)
}